package com.o2o.cleaning.month.platform.ebusiness_plat.mokuaixingxuan

import com.alibaba.fastjson.JSON
import com.mongodb.spark.MongoSpark
import com.o2o.utils.TimeStamp
import com.o2o.utils.times.TimesYearAll
import org.apache.commons.codec.digest.DigestUtils
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SparkSession}

object MoKuaiXingXuan {
  var platform = "mokuai" //平台名称
  var year = "2022" //当月的年份、月份
  var month = "9"
  var timeStamp = TimeStamp.TIME202209 //每个月固定时间戳
  //mongo库中的库名称
  var database = "Mokuai"
  //  mongo库中的集合名称
  var collection = "mokuai_detail2209"

  //原始数据
  var sourcePath = s"s3a://o2o-sourcedata-${year}/obs-source-${year}/${month}/${platform}/${collection}"

  //商品的清洗结果路径
  var resultPath = s"s3a://o2o-tempdata/zyf/${year}/${month}/${platform}/"
  var resultPath_dws = s"s3a://dws-data/split/split_data/${year}/${platform}/${month}/"

  def main(args: Array[String]): Unit = {

    //spark mongo连接配置
    val spark = SparkSession.builder()
      .master("local[*]")
      .config("spark.debug.maxToStringFields", "10000")
      .appName("MongoSparkConnectorIntro")
      .config("spark.mongodb.input.uri", "mongodb://root:O2Odata123!@ 114.115.144.192:27017/admin")
      .config("spark.mongodb.input.database", s"${database}")
      .config("spark.mongodb.input.collection", s"${collection}")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()
    //obs设置
    val sc: SparkContext = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("ERROR")

    //拉取MongoDB数据
    mongoExport(spark, sc, month, platform, database, collection)
    //清洗计算
    println(sourcePath + "=====")
    val frame2: DataFrame = mokuaiCaculate(spark, sourcePath)
      .where("sellCount>0").where("priceText>0").drop("add_to_field")
    frame2.registerTempTable("t2")
    println("销售量和销售额：")
    spark.sql(
      """
        |select count(1) count, cast(sum(sellCount) as bigInt) sellsum,cast(sum(salesAmount) as decimal(20,2)) salesum from t2
          """.stripMargin
    ).show()

    frame2.repartition(1).write.mode("overwrite").orc(resultPath)
    frame2.repartition(1).write.mode("overwrite").orc(resultPath_dws)

  }

  /**
    * 从mongo中拉取数据到OBS
    *
    * @param month
    * @param platform
    * @param collection
    */
  def mongoExport(spark: SparkSession, sc: SparkContext, month: String, platform: String, database: String, collection: String): Unit = {

    val values = MongoSpark.load(sc).map(line => {
      val nObject = JSON.parseObject(line.toJson())

      //所有字段转为string
      val keys = nObject.keySet().iterator()
      while (keys.hasNext) {
        var key = keys.next()
        nObject.put(key, nObject.get(key).toString)
      }

      nObject.remove("_id")
      nObject.toString
    })
    spark.read.json(values).repartition(1).write.mode("overwrite").orc(sourcePath)
  }

  /** *
    * 商品计算一：根据月销sellCount_monthly计算
    * 更改：根据总销量计算：
    */
  def mokuaiCaculate(spark: SparkSession, sourcePath: String): DataFrame = {
    val mokuaixingxuan = spark.read.orc(sourcePath).toJSON.rdd.map(line => {
      val nObject = JSON.parseObject(line)
      //处理品牌id
      val brandValueId = nObject.getOrDefault("brandId", "-1").toString
      val brandName = nObject.getOrDefault("brandName", "-1").toString
      if (brandValueId.equals("-1") && !brandName.equals("-1")) {
        //id -1  name apple /   -1  xiaomi  / -1  mi
        val brandValueId_1 = DigestUtils.md5Hex(brandName.toUpperCase())
        nObject.put("brandValueId", brandValueId_1)
      }
      /** *
        * 提取相关信息
        */
      val base = nObject.getOrDefault("Base_Info", "-1").toString

      val array = nObject.getJSONArray("add_to_field")
      val array_arr = nObject.getJSONArray("add_to_field").toArray()
      var sellCount = 0
      var priceText = 0.0
      var pricemid = 0.0

      //      try {
      // 长度为 1 的
      if (array.size() == 1) {
        //          val nObject1 = array.getJSONObject(0)
        //          priceText = checkOutPrice(nObject1.getString("priceText"))
        //          if (nObject1.getString("sellCount") != null && nObject1.getString("sellCount") != "") {
        //            sellCount += checkOutSellCount(nObject1.getString("sellCount"))
        //          }
      } else if (array.size() > 1) {
        // 长度大于 1 的
        /**
          * 销量用最后一天减第一天(根据时间戳降序取第一个减最后一个)
          * 价格取最低
          */
        val array_sort = array_arr.sortBy(x => {
          var crawl_date = JSON.parseObject(x.toString).get("crawl_date").toString
          //            if (crawl_date.contains(".")) {
          //              crawl_date = crawl_date.split(".")(0)
          //            }
          //            Integer.parseInt(crawl_date)
          crawl_date.toDouble
        })
        val lastArr = JSON.parseObject(array_sort.last.toString)
        val headArr = JSON.parseObject(array_sort.head.toString)

        sellCount = checkOutSellCount(lastArr.getString("sellCount")) - checkOutSellCount(headArr.getString("sellCount"))
        priceText = checkOutPrice(array.getJSONObject(0).getString("priceText"))
        /*for (i <- 0 to array.size() - 1) {
          val nObject1 = array.getJSONObject(i)

          // 一个月内最低的价格
          pricemid = checkOutPrice(nObject1.get("priceText").toString)
          if (i != 0 && (pricemid < priceText)) {
            priceText = pricemid
          } else if (i == 0) {
            priceText = pricemid
          }
        }*/
      }
      //      } catch {
      //        case e: Exception => println(e)
      //          println(nObject.getString("good_id") + "!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
      //      }


      val salesAmount = (priceText.toDouble * (sellCount)).toDouble.formatted("%.2f")

      nObject.put("sellCount", sellCount)
      nObject.put("Base_Info", base)
      nObject.put("priceText", priceText)
      nObject.put("salesAmount", salesAmount)
      nObject.remove("add_to_field")
      nObject.toString
    })
    spark.read.json(mokuaixingxuan)
  }

  /**
    * 此函数用来处理销量最后带W的
    *
    * @param sellcount
    * @return
    */
  def checkOutSellCount(sellcount: String): Int = {
    var midselld = 0.0
    var midselli = 0
    if (sellcount != null && sellcount != "") {
      if (sellcount.contains("W")) {
        midselld = sellcount.dropRight(1).toDouble
        midselli = (midselld * 10000).toInt
      } else {
        midselli = sellcount.toInt
      }
      midselli
    } else {
      0
    }
  }


  /**
    * 此函数用来处理价格中带-1 - , \ 的
    *
    * @param priceText
    * @return
    */
  def checkOutPrice(priceText: String): Double = {
    var price = priceText
    if (priceText.equals("-1")) {
      price = "0"
    } else if (priceText.contains("-")) {
      price = priceText.split("-")(1)
    } else if (priceText.contains(",")) {
      price = priceText.replace(",", "")
    } else if (priceText.contains("\"\"")) {
      price = priceText.split("\"\"")(0)
    } else {
      price
    }
    price.toDouble
  }
}
